package cn.com.bsfit.netty.server;

import cn.com.bsfit.netty.config.base.BaseConfigurable;
import cn.com.bsfit.netty.config.base.OSInfo;
import cn.com.bsfit.netty.handler.SocketReceiveDataHandler;
import cn.com.bsfit.netty.handler.SocketRequestHandler;
import cn.com.bsfit.netty.handler.SocketResponseHandler;
import cn.com.bsfit.netty.config.properties.BaseNettyServerProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 类名 NettySocketServer
 * 描述 Netty服务端类
 *
 * @author hedonglin
 * @version 1.0
 * @date 2019/9/29 10:24
 */
@Slf4j
@Configuration
@ConditionalOnProperty(prefix = "netty.server", name = "enable" ,havingValue = "true")
public class NettySocketServer implements ApplicationListener<ContextRefreshedEvent>, Runnable, Serializable, BaseConfigurable {
    private static final long serialVersionUID = -1114863053203847891L;

    @Qualifier(value = "startNettyServerPool")
    @Resource
    private ScheduledExecutorService startNettyServerPool;

    @Autowired(required = false)
    private BaseNettyServerProperties nettyServerProperties;

    @SuppressWarnings("all")
    @Autowired(required = false)
    @Qualifier(value = "workGroup")
    private EventLoopGroup workGroup;

    @SuppressWarnings("all")
    @Autowired(required = false)
    @Qualifier(value = "bossGroup")
    private EventLoopGroup bossGroup;

    @Autowired(required = false)
    private SocketResponseHandler responseHandler;

    @Autowired(required = false)
    private SocketRequestHandler requestHandler;

    @Override
    public void run() {
        try {
            doStart();
        } catch (Exception e) {
            //启动异常，关闭线程池
            log.error("{}", e);
            try {
                bossGroup.shutdownGracefully().sync();
                workGroup.shutdownGracefully().sync();
            } catch (InterruptedException e1) {
                log.error("{}", e1);
            }
        }
    }


    /**
     * Netty服务器启动业务代码
     */
    private void doStart() throws Exception {
        // 1、创建服务端启动类
        ServerBootstrap bootstrap = new ServerBootstrap();

        //2、设置启动参数
        BaseNettyServerProperties.RecBuf recbuf = nettyServerProperties.getRecbuf();
        bootstrap
                // 设置接收字节流大小空间
                .childOption(ChannelOption.RCVBUF_ALLOCATOR,
                        new AdaptiveRecvByteBufAllocator(recbuf.getMinSize(), recbuf.getInitial(), recbuf.getMaxSize()))
                //设置最大发送字节
                .childOption(ChannelOption.SO_SNDBUF, recbuf.getMaxSize())
                // 设置为TCP模式
                .childOption(ChannelOption.TCP_NODELAY, true)
                // 设置主线程和工作线程组
                .group(bossGroup, workGroup)
                // 根据系统设置Channel
                .channel(OSInfo.useNettyEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                // 设置服务端绑定端口
                .localAddress(new InetSocketAddress(nettyServerProperties.getPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                                // 响应处理器
                                .addLast(responseHandler)
                                // 字符串解码器
                                .addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8))
                                // 读取数据超时
                                .addLast(new ReadTimeoutHandler(nettyServerProperties.getRecbuf().getTimeout() * 1000, TimeUnit.MILLISECONDS))
                                // 写入数据超时
                                .addLast(new WriteTimeoutHandler(nettyServerProperties.getWriteBuf().getTimeout() * 1000, TimeUnit.MILLISECONDS))
                                // 数据聚合器(接受数据的处理)
                                .addLast(new SocketReceiveDataHandler())//接受数据处理器
                                // 业务请求处理器
                                .addLast("requestHandler", requestHandler);
                        ;
                    }
                });

        ChannelFuture future = bootstrap.bind().sync();

        //3、添加绑定端口监听器
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("Socket server bound successfully,the port is {}", nettyServerProperties.getPort());
                } else {
                    log.info("Socket server bound failed,the cause is {}", future.cause());
                }
            }
        });

        //4、添加关闭通道监听器
        future.channel().closeFuture().sync().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("the server channel is closed successfully");
                } else {
                    log.info("the server channel is closed with exception:{}", future.cause());
                }
            }
        });

    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        log.info("netty application starting");
        startNettyServerPool.schedule(this, nettyServerProperties.getStartDelayTime(), TimeUnit.MILLISECONDS);
    }


//    @PostConstruct
//    public void init(){
//        System.out.println("netty application starting");
//    }


    //销毁方法
    @PreDestroy
    public void stop() {
        try {
            //boss线程组关闭监听器
            bossGroup.shutdownGracefully().sync().addListener(new GenericFutureListener<Future<Object>>() {
                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        log.info("the bossGroup is closed successfully");
                    } else {
                        log.info("the bossGroup is closed with exception:{}", future.cause());
                    }
                }
            });
            //worker线程组关闭监听器
            workGroup.shutdownGracefully().sync().addListener(new GenericFutureListener<Future<Object>>() {
                @Override
                public void operationComplete(Future<Object> future) throws Exception {
                    if (future.isSuccess()) {
                        log.info("the workGroup is closed successfully");
                    } else {
                        log.info("the workGroup is closed with exception:{}", future.cause());
                    }
                }
            });
        } catch (Exception e) {
            log.error("{}", e);
        }
    }
}
